package monitor

import (
	"context"
	"encoding/json"
	"fmt"
	gorillaWs "github.com/gorilla/websocket"
	"github.com/kataras/iris/v12/websocket"
	"github.com/kataras/neffos"
	"github.com/kataras/neffos/gorilla"
	"github.com/olivere/elastic/v7"
	"gluenet/modules/manager/stream/handler"
	apis2 "gluenet/pkg/apis"
	"gluenet/pkg/meta"
	"gluenet/pkg/ops"
	"net/http"
	"reflect"
	"time"
)

//const (
//	info = `
//from(bucket: "data")
//    |> range(start: %v, stop: %v)
//    |> filter(fn: (r) => r["_measurement"] == "logs")
//    |> filter(fn: (r) => r["guid"] == "%v")
//    |> drop(columns: ["file","container_name","namespace","namespaces_labels","pod_annotations","pod_ips","pod_labels","stream","pod_name","pod_uid","node"])
//    |> window(every: %v)
//    |> count()
//`
//	logs = `
//from(bucket: "data")
//    |> range(start: %v, stop: %v)
//    |> filter(fn: (r) => r["_measurement"] == "logs")
//    |> filter(fn: (r) => r["guid"] == "%v")
//    |> drop(columns: ["file","container_name","namespace","namespaces_labels","pod_annotations","pod_ips","pod_labels","stream","pod_name","pod_uid","node"])
//    |> group(columns: ["_result"])
//    |> sort(columns: ["_time"])
//`
//)

func LogsCounts(c *apis2.GlueContext) {
	startTime := c.WebContext.URLParam("start")
	guid := c.WebContext.URLParam("guid")
	//platform := c.WebContext.URLParam("platform")
	container := c.WebContext.URLParam("container")
	endTime := c.WebContext.URLParam("end")
	bucket := c.WebContext.URLParam("bucket")

	//guid = "default"
	//c.Logger.Debugf("start %v end: %v platform: %v container: %v guid: %v", startTime, endTime, platform, container, guid)

	//c.Logger.Debugf("end: %v", time.Now().Unix())
	var ins *meta.Instance
	if err := c.DataBase.InstanceGet(guid, &ins); err != nil {
		c.Logger.Errorf("query instance error :%v", err)
		c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
		return
	}

	var data *meta.Logs
	if err := c.DataBase.LogsCounts(ins.Platform, guid, container, startTime, endTime, bucket, &data); err != nil {
		c.Logger.Errorf("query error :%v", err)
		c.WebContext.JSON(apis2.ResponseReqErr(err, data))
		return
	}

	//c.Logger.Debugf("end: %v", time.Now().Unix())
	c.WebContext.JSON(apis2.ResponseOK(data.LogBuckets))
}

func LogsInfo(c *apis2.GlueContext) {
	startTime := c.WebContext.URLParam("start")
	guid := c.WebContext.URLParam("guid")
	//platform := c.WebContext.URLParam("platform")
	container := c.WebContext.URLParam("container")
	endTime := c.WebContext.URLParam("end")

	//guid = "default"
	var ins *meta.Instance
	if err := c.DataBase.InstanceGet(guid, &ins); err != nil {
		c.Logger.Errorf("query instance error :%v", err)
		c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
		return
	}
	//if len(ins) == 0 {
	//	c.Logger.Errorf("instance not exist")
	//	c.WebContext.JSON(apis.ResponseReqErr(fmt.Errorf("instance not exist"), nil))
	//	return
	//}

	//c.Logger.Debugf("start %v end: %v platform: %v container: %v guid: %v", startTime, endTime, platform, container, guid)

	var data *meta.Logs
	if err := c.DataBase.LogsQuery(ins.Platform, guid, container, startTime, endTime, &data); err != nil {
		c.Logger.Errorf("query error :%v", err)
		c.WebContext.JSON(apis2.ResponseReqErr(err, data.Logs))
		return
	}

	c.Logger.Debugf("end: %v", time.Now().Unix())
	c.WebContext.JSON(apis2.ResponseOK(data.Logs))
}

func LogSocket(c *apis2.GlueContext) {
	guid := c.WebContext.URLParam("guid")
	//platform := c.WebContext.URLParam("platform")
	container := c.WebContext.URLParam("container")
	//guid = "default"

	var ins *meta.Instance
	if err := c.DataBase.InstanceGet(guid, &ins); err != nil {
		c.Logger.Errorf("query instance error :%v", err)
		c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
		return
	}
	//if len(ins) == 0 {
	//	c.Logger.Errorf("instance not exist")
	//	c.WebContext.JSON(apis.ResponseReqErr(fmt.Errorf("instance not exist"), nil))
	//	return
	//}

	ctx, cancel := context.WithCancel(c.Context)
	var ActuallyNodeCpu = websocket.New(gorilla.Upgrader(gorillaWs.Upgrader{
		CheckOrigin: func(*http.Request) bool { return true },
	}), websocket.Events{
		websocket.OnNativeMessage: func(conn *neffos.NSConn, message neffos.Message) error { return nil },
	})

	ActuallyNodeCpu.OnConnect = func(n *websocket.Conn) error {
		c.Logger.Debugf("OnConnect from server %v", n.ID())
		tk := time.NewTicker(time.Second * 5)
		for {
			select {
			case <-tk.C:
				end := time.Now().Unix()
				start := end - 4
				//sql := fmt.Sprintf(logs, start, end, guid)
				//c.Logger.Debugf("log socket sql %v", sql)

				startTime := fmt.Sprintf("%v", start)
				endTime := fmt.Sprintf("%v", end)
				var data *meta.Logs
				if err := c.DataBase.LogsQuery(ins.Platform, guid, container, startTime, endTime, &data); err != nil {
					c.Logger.Errorf("query error :%v", err)
					c.WebContext.JSON(apis2.ResponseReqErr(err, data))
				}

				bts, _ := json.Marshal(data.Logs)
				n.Write(neffos.Message{
					Body:     bts,
					IsNative: true,
				})
			case <-ctx.Done():
				c.Logger.Debugf("push down %v", n.ID())
				return nil
			}
		}
		return nil
	}

	ActuallyNodeCpu.OnDisconnect = func(n *websocket.Conn) {
		cancel()
		c.Logger.Debugf("Disconnected from server", n.ID())
	}

	ActuallyNodeCpu.OnUpgradeError = func(err error) {
		cancel()
		c.Logger.Debugf("OnUpgradeError from server %v", err)
	}

	websocket.Handler(ActuallyNodeCpu)(c.WebContext)
}

// LogsCount 日志统计
func LogsCount(c *apis2.GlueContext) {
	//Get es client index
	client := c.DataBase.GetESClient()
	//Get url parameter of request
	startTime := c.WebContext.URLParam("start")
	guid := c.WebContext.URLParam("guid")
	container := c.WebContext.URLParam("container")
	endTime := c.WebContext.URLParam("end")
	bucket := c.WebContext.URLParam("bucket")
	//Check if instance exist
	var ins *meta.Instance
	if err := c.DataBase.InstanceGet(guid, &ins); err != nil {
		c.Logger.Errorf("query instance error :", err.Error())
		c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
		return
	}
	//1.Query from db
	//Set criteria for the query
	boolQuery := elastic.NewBoolQuery()
	termQueryGUID := elastic.NewQueryStringQuery("application_guid:" + guid)
	termQueryContainer := elastic.NewQueryStringQuery("container:" + container)
	rangeQueryTimeStamp := elastic.NewRangeQuery("timestamp").Gte(startTime).Lte(endTime)
	boolQuery.Must(termQueryGUID, termQueryContainer, rangeQueryTimeStamp)
	//Set the aggregation with interval of timestamp and term of level
	interval, _ := time.ParseDuration(bucket)
	timeStamp := elastic.NewHistogramAggregation().Field("timestamp").Interval(interval.Seconds()).MinDocCount(0)
	histogram := elastic.NewTermsAggregation().Field("level").MinDocCount(0)
	timeStamp.SubAggregation("level", histogram)
	//Start query
	searchResult, err := client.Search().
		Index(ops.EsLog).
		Query(boolQuery).
		Aggregation("timestamp", timeStamp).
		Size(0).
		Do(context.Background())
	if err != nil {
		c.Logger.Errorf("query error :", err.Error())
		c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
		return
	}
	//2.Data organization
	aggResByTimeStamp, found := searchResult.Aggregations.Histogram("timestamp")
	if !found {
		c.Logger.Errorf("no aggregate results found with timestamp,error :", err.Error())
		c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
		return
	}
	var res []map[string]interface{}
	for _, bucket := range aggResByTimeStamp.Buckets {
		bucketValue := bucket.Key
		aggResByLevel, found := bucket.Aggregations.Terms("level")
		if !found {
			c.Logger.Errorf("no aggregate results found with level,error :", err.Error())
			c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
			return
		}
		for _, log := range aggResByLevel.Buckets {
			res = append(res, map[string]interface{}{
				"counts": log.DocCount,
				"label":  "logs",
				"start":  int(bucketValue),
				"stop":   int(bucketValue + interval.Seconds()),
				"type":   log.Key,
			})
		}
	}
	//3.Response
	c.WebContext.JSON(apis2.ResponseOK(res))
}

// LogsInfoList 日志信息
func LogsInfoList(c *apis2.GlueContext) {
	//Get es client index
	client := c.DataBase.GetESClient()
	//Get url parameter of request
	startTime := c.WebContext.URLParam("start")
	guid := c.WebContext.URLParam("guid")
	container := c.WebContext.URLParam("container")
	endTime := c.WebContext.URLParam("end")
	//Check if instance exist
	var ins *meta.Instance
	if err := c.DataBase.InstanceGet(guid, &ins); err != nil {
		c.Logger.Errorf("query instance error :", err.Error())
		c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
		return
	}
	//1.Query from db
	//Set criteria for the query
	boolQuery := elastic.NewBoolQuery()
	termQueryGUID := elastic.NewQueryStringQuery("application_guid:" + guid)
	termQueryContainer := elastic.NewQueryStringQuery("container:" + container)
	rangeQueryTimeStamp := elastic.NewRangeQuery("timestamp").Gte(startTime).Lte(endTime)
	boolQuery.Must(termQueryGUID, termQueryContainer, rangeQueryTimeStamp)
	//Start query
	searchResult, err := client.Search().
		Index(ops.EsLog).
		Query(boolQuery).
		Sort("timestamp", true).
		Do(context.Background())
	if err != nil {
		c.Logger.Errorf("query error :", err.Error())
		c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
		return
	}
	//2.Data organization
	var res []map[string]interface{}
	if searchResult.TotalHits() > 0 {
		var esLog handler.EsLog
		for _, item := range searchResult.Each(reflect.TypeOf(esLog)) {
			if t, ok := item.(handler.EsLog); ok {
				res = append(res, map[string]interface{}{
					"label":   "logs",
					"message": t.Message,
					"time":    t.Timestamp,
					"type":    t.Level,
				})
			}
		}
	}
	//3.Response
	c.WebContext.JSON(apis2.ResponseOK(res))
}

// LogsSocket 日志Socket
func LogsSocket(c *apis2.GlueContext) {
	//Get es client index
	client := c.DataBase.GetESClient()
	guid := c.WebContext.URLParam("guid")
	container := c.WebContext.URLParam("container")
	//Check if instance exist
	var ins *meta.Instance
	if err := c.DataBase.InstanceGet(guid, &ins); err != nil {
		c.Logger.Errorf("query instance error :", err.Error())
		c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
		return
	}
	//1.Create websocket server
	ctx, cancel := context.WithCancel(c.Context)
	var websocketServer = websocket.New(gorilla.Upgrader(gorillaWs.Upgrader{
		CheckOrigin: func(*http.Request) bool { return true },
	}), websocket.Events{
		websocket.OnNativeMessage: func(conn *neffos.NSConn, message neffos.Message) error { return nil },
	})
	websocketServer.OnConnect = func(n *websocket.Conn) error {
		c.Logger.Debugf("OnConnect from server %v", n.ID())
		tk := time.NewTicker(time.Second * 5)
		for {
			select {
			case <-tk.C:
				endTime := time.Now().Unix()
				startTime := endTime - 4
				//2.Query from db
				//Set criteria for the query
				boolQuery := elastic.NewBoolQuery()
				termQueryGUID := elastic.NewQueryStringQuery("application_guid:" + guid)
				termQueryContainer := elastic.NewQueryStringQuery("container:" + container)
				rangeQueryTimeStamp := elastic.NewRangeQuery("timestamp").Gte(startTime).Lte(endTime)
				boolQuery.Must(termQueryGUID, termQueryContainer, rangeQueryTimeStamp)
				//Start query
				searchResult, err := client.Search().
					Index(ops.EsLog).
					Query(boolQuery).
					Sort("timestamp", true).
					Do(context.Background())
				if err != nil {
					c.Logger.Errorf("query error :", err.Error())
					c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
				}
				//3.Data organization
				var res []map[string]interface{}
				if searchResult.TotalHits() > 0 {
					var esLog handler.EsLog
					for _, item := range searchResult.Each(reflect.TypeOf(esLog)) {
						if t, ok := item.(handler.EsLog); ok {
							res = append(res, map[string]interface{}{
								"label":   "logs",
								"message": t.Message,
								"time":    t.Timestamp,
								"type":    t.Level,
							})
						}
					}
				}
				//4.Send data to client
				bts, _ := json.Marshal(res)
				n.Write(neffos.Message{
					Body:     bts,
					IsNative: true,
				})
			case <-ctx.Done():
				c.Logger.Debugf("push down %v", n.ID())
				return nil
			}
		}
	}
	websocketServer.OnDisconnect = func(n *websocket.Conn) {
		cancel()
		c.Logger.Debugf("Disconnected from server", n.ID())
	}
	websocketServer.OnUpgradeError = func(err error) {
		cancel()
		c.Logger.Debugf("OnUpgradeError from server %v", err)
	}
	websocket.Handler(websocketServer)(c.WebContext)
}
